/*
 *
 *  *  Copyright 2010-2016 OrientDB LTD (http://orientdb.com)
 *  *
 *  *  Licensed under the Apache License, Version 2.0 (the "License");
 *  *  you may not use this file except in compliance with the License.
 *  *  You may obtain a copy of the License at
 *  *
 *  *       http://www.apache.org/licenses/LICENSE-2.0
 *  *
 *  *  Unless required by applicable law or agreed to in writing, software
 *  *  distributed under the License is distributed on an "AS IS" BASIS,
 *  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  *  See the License for the specific language governing permissions and
 *  *  limitations under the License.
 *  *
 *  * For more information: http://orientdb.com
 *
 */
package com.orientechnologies.orient.server.distributed.impl;

import static com.orientechnologies.orient.core.config.OGlobalConfiguration.DISTRIBUTED_MAX_STARTUP_DELAY;

import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.HazelcastInstanceNotActiveException;
import com.hazelcast.core.Member;
import com.orientechnologies.common.concur.OOfflineNodeException;
import com.orientechnologies.common.concur.lock.OInterruptedException;
import com.orientechnologies.common.console.OConsoleReader;
import com.orientechnologies.common.console.ODefaultConsoleReader;
import com.orientechnologies.common.exception.OException;
import com.orientechnologies.common.io.OFileUtils;
import com.orientechnologies.common.io.OIOException;
import com.orientechnologies.common.log.OAnsiCode;
import com.orientechnologies.common.log.OLogManager;
import com.orientechnologies.common.parser.OSystemVariableResolver;
import com.orientechnologies.common.util.OArrays;
import com.orientechnologies.common.util.OCallable;
import com.orientechnologies.orient.core.OConstants;
import com.orientechnologies.orient.core.OSignalHandler;
import com.orientechnologies.orient.core.Orient;
import com.orientechnologies.orient.core.command.OCommandDistributedReplicateRequest;
import com.orientechnologies.orient.core.config.OContextConfiguration;
import com.orientechnologies.orient.core.config.OGlobalConfiguration;
import com.orientechnologies.orient.core.db.ODatabaseDocumentInternal;
import com.orientechnologies.orient.core.db.ODatabaseLifecycleListener;
import com.orientechnologies.orient.core.db.ODatabaseRecordThreadLocal;
import com.orientechnologies.orient.core.db.OScenarioThreadLocal;
import com.orientechnologies.orient.core.db.OSystemDatabase;
import com.orientechnologies.orient.core.db.OrientDBConfig;
import com.orientechnologies.orient.core.db.OrientDBInternal;
import com.orientechnologies.orient.core.exception.OConfigurationException;
import com.orientechnologies.orient.core.exception.ODatabaseException;
import com.orientechnologies.orient.core.exception.OSecurityAccessException;
import com.orientechnologies.orient.core.metadata.schema.OClass;
import com.orientechnologies.orient.core.metadata.schema.OSchema;
import com.orientechnologies.orient.core.metadata.security.OSecurityUser;
import com.orientechnologies.orient.core.record.impl.ODocument;
import com.orientechnologies.orient.core.transaction.ONodeId;
import com.orientechnologies.orient.core.tx.OTxMetadataHolder;
import com.orientechnologies.orient.core.tx.OTxMetadataHolderImpl;
import com.orientechnologies.orient.distributed.ONodeConfig;
import com.orientechnologies.orient.distributed.ONodeListenerConfig;
import com.orientechnologies.orient.distributed.db.OrientDBDistributed;
import com.orientechnologies.orient.server.OServer;
import com.orientechnologies.orient.server.config.OServerConfiguration;
import com.orientechnologies.orient.server.config.OServerHandlerConfiguration;
import com.orientechnologies.orient.server.config.OServerParameterConfiguration;
import com.orientechnologies.orient.server.distributed.NODE_STATUS;
import com.orientechnologies.orient.server.distributed.ODistributedConfiguration;
import com.orientechnologies.orient.server.distributed.ODistributedException;
import com.orientechnologies.orient.server.distributed.ODistributedLifecycleListener;
import com.orientechnologies.orient.server.distributed.ODistributedMessageService;
import com.orientechnologies.orient.server.distributed.ODistributedRequest;
import com.orientechnologies.orient.server.distributed.ODistributedRequest.EXECUTION_MODE;
import com.orientechnologies.orient.server.distributed.ODistributedRequestId;
import com.orientechnologies.orient.server.distributed.ODistributedResponse;
import com.orientechnologies.orient.server.distributed.ODistributedResponseManager;
import com.orientechnologies.orient.server.distributed.ODistributedResponseManagerFactory;
import com.orientechnologies.orient.server.distributed.ODistributedResponseManagerImpl;
import com.orientechnologies.orient.server.distributed.ODistributedServerManager;
import com.orientechnologies.orient.server.distributed.ODistributedStartupException;
import com.orientechnologies.orient.server.distributed.ODistributedStrategy;
import com.orientechnologies.orient.server.distributed.OLoggerDistributed;
import com.orientechnologies.orient.server.distributed.OModifiableDistributedConfiguration;
import com.orientechnologies.orient.server.distributed.ORemoteServerAvailabilityCheck;
import com.orientechnologies.orient.server.distributed.ORemoteServerController;
import com.orientechnologies.orient.server.distributed.ORemoteTaskFactoryManager;
import com.orientechnologies.orient.server.distributed.config.OClusterConfiguration;
import com.orientechnologies.orient.server.distributed.impl.metadata.OClassDistributed;
import com.orientechnologies.orient.server.distributed.impl.task.ODropDatabaseTask;
import com.orientechnologies.orient.server.distributed.impl.task.ONewDeltaTaskResponse;
import com.orientechnologies.orient.server.distributed.impl.task.ORemoteTaskFactoryManagerImpl;
import com.orientechnologies.orient.server.distributed.impl.task.ORestartServerTask;
import com.orientechnologies.orient.server.distributed.impl.task.OStopServerTask;
import com.orientechnologies.orient.server.distributed.impl.task.OSyncDatabaseNewDeltaTask;
import com.orientechnologies.orient.server.distributed.impl.task.OSyncDatabaseTask;
import com.orientechnologies.orient.server.distributed.impl.task.OUpdateDatabaseConfigurationTask;
import com.orientechnologies.orient.server.distributed.task.OAbstractRemoteTask;
import com.orientechnologies.orient.server.distributed.task.ODatabaseIsOldException;
import com.orientechnologies.orient.server.distributed.task.ODistributedDatabaseDeltaSyncException;
import com.orientechnologies.orient.server.distributed.task.ODistributedLockException;
import com.orientechnologies.orient.server.distributed.task.ORemoteTask;
import com.orientechnologies.orient.server.hazelcast.OHazelcastClusterMetadataManager;
import com.orientechnologies.orient.server.network.OServerNetworkListener;
import com.orientechnologies.orient.server.plugin.OServerPluginAbstract;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.AtomicMoveNotSupportedException;
import java.nio.file.DirectoryNotEmptyException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardCopyOption;
import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.TimerTask;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import sun.misc.Signal;

/**
 * Plugin to manage the distributed environment.
 *
 * @author Luca Garulli (l.garulli--at--orientechnologies.com)
 */
public class ODistributedPlugin extends OServerPluginAbstract implements ODistributedServerManager {
  private static final OLoggerDistributed logger =
      OLoggerDistributed.logger(ODistributedPlugin.class);
  public static final String REPLICATOR_USER = "_CrossServerTempUser";

  protected static final String PAR_DEF_DISTRIB_DB_CONFIG = "configuration.db.default";
  protected static final String NODE_NAME_ENV = "ORIENTDB_NODE_NAME";

  private OServer serverInstance;
  private String nodeName = null;
  protected File defaultDatabaseConfigFile;
  protected List<ODistributedLifecycleListener> listeners = new ArrayList<>();
  protected ORemoteServerManager remoteServerManager;

  // LOCAL MSG COUNTER
  protected AtomicLong localMessageIdCounter = new AtomicLong();

  protected static final int DEPLOY_DB_MAX_RETRIES = 10;
  protected Set<String> installingDatabases =
      Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>());
  protected volatile ODistributedMessageServiceImpl messageService;
  protected Date startedOn = new Date();
  protected ODistributedStrategy responseManagerFactory = new ODefaultDistributedStrategy();
  protected ORemoteTaskFactoryManager taskFactoryManager = new ORemoteTaskFactoryManagerImpl(this);

  private volatile String lastServerDump = "";
  protected CountDownLatch serverStarted = new CountDownLatch(1);

  private TimerTask haStatsTask = null;
  private TimerTask healthCheckerTask = null;
  protected OSignalHandler.OSignalListener signalListener;

  private final OHazelcastClusterMetadataManager clusterManager;

  public ODistributedPlugin() {
    clusterManager = new OHazelcastClusterMetadataManager(this);
  }

  public void waitUntilNodeOnline() throws InterruptedException {
    serverStarted.await();
  }

  public void waitUntilNodeOnline(final String nodeName, final String databaseName)
      throws InterruptedException {
    while (getDatabase(databaseName) == null || !isNodeOnline(nodeName, databaseName))
      Thread.sleep(100);
  }

  @Override
  public void config(OServer oServer, OServerParameterConfiguration[] iParams) {
    serverInstance = oServer;
    oServer.setVariable("ODistributedAbstractPlugin", this);

    for (OServerParameterConfiguration param : iParams) {
      if (param.name.equalsIgnoreCase("enabled")) {
        if (!Boolean.parseBoolean(OSystemVariableResolver.resolveSystemVariables(param.value))) {
          // DISABLE IT
          enabled = false;
          return;
        }
      } else if (param.name.equalsIgnoreCase("nodeName")) {
        nodeName = param.value;
        if (nodeName.contains("."))
          throw new OConfigurationException(
              "Illegal node name '" + nodeName + "'. '.' is not allowed in node name");
      } else if (param.name.startsWith(PAR_DEF_DISTRIB_DB_CONFIG)) {
        setDefaultDatabaseConfigFile(param.value);
      }
    }

    this.remoteServerManager =
        new ORemoteServerManager(
            nodeName,
            new ORemoteServerAvailabilityCheck() {
              @Override
              public boolean isNodeAvailable(String node) {
                return ODistributedPlugin.this.isNodeAvailable(node);
              }

              @Override
              public void nodeDisconnected(String node) {
                ODistributedPlugin.this.removeServer(node, true);
              }
            });
    if (nodeName == null) assignNodeName();
    clusterManager.configHazelcastPlugin(oServer, iParams, nodeName);
  }

  public ORemoteServerManager getRemoteServerManager() {
    return remoteServerManager;
  }

  public File getDefaultDatabaseConfigFile() {
    return defaultDatabaseConfigFile;
  }

  @Override
  public <T> T executeInDistributedDatabaseLock(
      String databaseName,
      long timeoutLocking,
      OModifiableDistributedConfiguration lastCfg,
      OCallable<T, OModifiableDistributedConfiguration> iCallback) {
    return clusterManager.executeInDistributedDatabaseLock(
        databaseName, timeoutLocking, lastCfg, iCallback);
  }

  public <T> T executeInDistributedDatabaseLock(
      String databaseName, long timeoutLocking, Callable<T> iCallback) {
    return clusterManager.executeInDistributedDatabaseLock(databaseName, timeoutLocking, iCallback);
  }

  @Override
  public boolean isWriteQuorumPresent(String databaseName) {
    return clusterManager.isWriteQuorumPresent(databaseName);
  }

  public void setDefaultDatabaseConfigFile(final String iFile) {
    defaultDatabaseConfigFile = new File(OSystemVariableResolver.resolveSystemVariables(iFile));
    if (!defaultDatabaseConfigFile.exists())
      throw new OConfigurationException(
          "Cannot find distributed database config file: " + defaultDatabaseConfigFile);
  }

  @Override
  public void startup() {
    if (!enabled) return;
    if (serverInstance.getDatabases() instanceof OrientDBDistributed)
      ((OrientDBDistributed) serverInstance.getDatabases()).setPlugin(this);

    // REGISTER TEMPORARY USER FOR REPLICATION PURPOSE
    serverInstance.addTemporaryUser(REPLICATOR_USER, "" + new SecureRandom().nextLong(), "*");

    // CLOSE ALL CONNECTIONS TO THE SERVERS
    remoteServerManager.closeAll();

    try {
      clusterManager.startupHazelcastPlugin();

      OContextConfiguration ctx = serverInstance.getContextConfiguration();
      final long statsDelay = ctx.getValueAsLong(OGlobalConfiguration.DISTRIBUTED_DUMP_STATS_EVERY);
      if (statsDelay > 0) {
        haStatsTask =
            new TimerTask() {
              @Override
              public void run() {
                ODistributedPlugin.this.dumpStats();
              }
            };
        serverInstance.getDatabases().schedule(haStatsTask, statsDelay, statsDelay);
      }

      final long healthChecker =
          ctx.getValueAsLong(OGlobalConfiguration.DISTRIBUTED_CHECK_HEALTH_EVERY);
      if (healthChecker > 0) {
        OClusterHealthChecker checkTask = new OClusterHealthChecker(this, healthChecker);
        healthCheckerTask =
            new TimerTask() {
              @Override
              public void run() {
                serverInstance.getDatabases().execute(checkTask);
              }
            };
        serverInstance.getDatabases().schedule(healthCheckerTask, healthChecker, healthChecker);
      }

      signalListener =
          new OSignalHandler.OSignalListener() {
            @Override
            public void onSignal(final Signal signal) {
              if (signal.toString().trim().equalsIgnoreCase("SIGTRAP")) dumpStats();
            }
          };
      Orient.instance().getSignalHandler().registerListener(signalListener);
    } catch (Exception e) {
      logger.errorNode(nodeName, "Error on starting distributed plugin", e);
      throw OException.wrapException(
          new ODistributedStartupException("Error on starting distributed plugin"), e);
    }

    dumpServersStatus();
  }

  @Override
  public ODistributedPlugin registerLifecycleListener(
      final ODistributedLifecycleListener iListener) {
    if (iListener == null) {
      throw new NullPointerException();
    }
    listeners.add(iListener);
    return this;
  }

  @Override
  public ODistributedPlugin unregisterLifecycleListener(
      final ODistributedLifecycleListener iListener) {
    listeners.remove(iListener);
    return this;
  }

  @Override
  public void shutdown() {
    if (!enabled) return;
    OSignalHandler signalHandler = Orient.instance().getSignalHandler();
    if (signalHandler != null) signalHandler.unregisterListener(signalListener);

    logger.warnNode(nodeName, "Shutting down node '%s'...", nodeName);
    setNodeStatus(NODE_STATUS.SHUTTINGDOWN);

    clusterManager.prepareHazelcastPluginShutdown();
    try {
      if (healthCheckerTask != null) healthCheckerTask.cancel();
      if (haStatsTask != null) haStatsTask.cancel();

      // CLOSE ALL CONNECTIONS TO THE SERVERS
      remoteServerManager.closeAll();

      setNodeStatus(NODE_STATUS.OFFLINE);

    } catch (HazelcastInstanceNotActiveException e) {
      // HZ IS ALREADY DOWN, IGNORE IT
    }
    clusterManager.hazelcastPluginShutdown();
  }

  public void removeDbFromClusterMetadata(String name) {
    clusterManager.removeDbFromClusterMetadata(name);
  }

  public void dropOnAllServers(final String dbName) {
    Set<String> servers = clusterManager.dropDbFromConfiguration(dbName);
    if (!servers.isEmpty() && getDatabase(dbName) != null) {
      sendRequest(dbName, servers, new ODropDatabaseTask());
    }
  }

  public void dropConfig(String dbName) {
    clusterManager.dropDatabaseConfiguration(dbName);
  }

  @Override
  public String getName() {
    return "cluster";
  }

  @Override
  public void sendShutdown() {
    shutdown();
  }

  public OServer getServerInstance() {
    return serverInstance;
  }

  @Override
  public ONodeConfig getLocalNodeConfiguration() {
    ONodeConfig nodeCfg = new ONodeConfig();

    nodeCfg.setId(getLocalNodeId());
    nodeCfg.setUuid(clusterManager.getLocalNodeUuid());
    nodeCfg.setName(nodeName);
    nodeCfg.setVersion(OConstants.getRawVersion());
    nodeCfg.setPublicAddress(clusterManager.getPublicAddress());
    nodeCfg.setStartedOn(startedOn);
    nodeCfg.setStatus(getNodeStatus().toString());
    nodeCfg.setConnections(serverInstance.getClientConnectionManager().getTotal());

    List<ONodeListenerConfig> listeners = new ArrayList<>();
    for (OServerNetworkListener listener : serverInstance.getNetworkListeners()) {
      listeners.add(
          new ONodeListenerConfig(
              listener.getProtocolType().getSimpleName(), listener.getListeningAddress(true)));
    }
    nodeCfg.setListeners(listeners);

    // STORE THE TEMP USER/PASSWD USED FOR REPLICATION
    final OSecurityUser user = serverInstance.getSecurity().getUser(REPLICATOR_USER);
    if (user != null)
      nodeCfg.setReplicator(serverInstance.getSecurity().getUser(REPLICATOR_USER).getPassword());

    nodeCfg.setDatabases(getManagedDatabases());

    final long maxMem = Runtime.getRuntime().maxMemory();
    final long totMem = Runtime.getRuntime().totalMemory();
    final long freeMem = Runtime.getRuntime().freeMemory();
    final long usedMem = totMem - freeMem;

    nodeCfg.setUsedMemory(usedMem);
    nodeCfg.setFreeMemory(freeMem);
    nodeCfg.setMaxMemory(maxMem);

    nodeCfg.setLatencies("latencies", getMessageService().getLatencies());
    nodeCfg.setMessages("messages", getMessageService().getMessageStats());

    for (Iterator<ODatabaseLifecycleListener> it = Orient.instance().getDbLifecycleListeners();
        it.hasNext(); ) {
      final ODatabaseLifecycleListener listener = it.next();
      if (listener != null) listener.onLocalNodeConfigurationRequest(nodeCfg);
    }

    return nodeCfg;
  }

  @Override
  public ODistributedConfiguration getDatabaseConfiguration(String iDatabaseName) {
    return clusterManager.getDatabaseConfiguration(iDatabaseName);
  }

  public boolean isEnabled() {
    return enabled;
  }

  @Override
  public ODistributedResponse sendRequest(
      final String iDatabaseName, final Collection<String> iTargetNodes, final ORemoteTask iTask) {
    return sendRequest(iDatabaseName, iTargetNodes, iTask, nextRequestId(), null, null);
  }

  @Override
  public ODistributedResponse sendSingleRequest(
      String databaseName, String node, ORemoteTask iTask) {
    return sendRequest(
        databaseName, Collections.singletonList(node), iTask, nextRequestId(), null, null);
  }

  public ODistributedResponse sendRequest(
      final String iDatabaseName,
      final Collection<String> iTargetNodes,
      final ORemoteTask iTask,
      final ODistributedRequestId reqId,
      final Object localResult,
      ODistributedResponseManagerFactory responseManagerFactory) {
    final ODistributedRequest.EXECUTION_MODE iExecutionMode = EXECUTION_MODE.RESPONSE;
    final ODistributedRequest req = new ODistributedRequest(this, reqId, iDatabaseName, iTask);

    if (iTargetNodes == null || iTargetNodes.isEmpty()) {
      logger.errorOut(
          nodeName, null, "No nodes configured for partition '%s' request: %s", iDatabaseName, req);
      throw new ODistributedException(
          "No nodes configured '" + iDatabaseName + "' request: " + req);
    }

    getMessageService().updateMessageStats(iTask.getName());
    if (responseManagerFactory != null) {
      return send2Nodes(req, iTargetNodes, iExecutionMode, localResult, responseManagerFactory);
    } else {
      return send2Nodes(req, iTargetNodes, iExecutionMode, localResult);
    }
  }

  protected void checkForServerOnline(final ODistributedRequest iRequest)
      throws ODistributedException {
    final NODE_STATUS srvStatus = getNodeStatus();
    if (srvStatus == NODE_STATUS.OFFLINE || srvStatus == NODE_STATUS.SHUTTINGDOWN) {
      logger.errorOut(
          this.nodeName,
          null,
          "Local server is not online (status='%s'). Request %s will be ignored",
          srvStatus,
          iRequest);
      throw new OOfflineNodeException(
          "Local server is not online (status='"
              + srvStatus
              + "'). Request "
              + iRequest
              + " will be ignored");
    }
  }

  public ODistributedResponse send2Nodes(
      final ODistributedRequest iRequest,
      Collection<String> iNodes,
      final ODistributedRequest.EXECUTION_MODE iExecutionMode,
      final Object localResult,
      ODistributedResponseManagerFactory responseManagerFactory) {
    try {
      checkForServerOnline(iRequest);

      final String databaseName = iRequest.getDatabaseName();

      if (iNodes.isEmpty()) {
        logger.errorOut(
            this.nodeName,
            null,
            "No nodes configured for database '%s' request: %s",
            databaseName,
            iRequest);
        throw new ODistributedException(
            "No nodes configured for partition '" + databaseName + "' request: " + iRequest);
      }
      final ORemoteTask task = iRequest.getTask();
      final boolean checkNodesAreOnline = task.isNodeOnlineRequired();

      final ODistributedConfiguration cfg;
      final Set<String> nodesConcurToTheQuorum;
      int availableNodes = iNodes.size();
      int onlineMasters;
      if (databaseName != null) {
        cfg = getDatabaseConfiguration(databaseName);
        nodesConcurToTheQuorum =
            getDistributedStrategy().getNodesConcurInQuorum(this, cfg, iRequest, iNodes);

        // AFTER COMPUTED THE QUORUM, REMOVE THE OFFLINE NODES TO HAVE THE LIST OF REAL AVAILABLE
        // NODES

        if (checkNodesAreOnline) {
          availableNodes =
              getNodesWithStatus(
                  iNodes,
                  databaseName,
                  ODistributedServerManager.DB_STATUS.ONLINE,
                  ODistributedServerManager.DB_STATUS.BACKUP,
                  ODistributedServerManager.DB_STATUS.SYNCHRONIZING);
        }

        // all online masters
        onlineMasters =
            getOnlineNodes(databaseName).stream()
                .filter(f -> cfg.getServerRole(f) == ODistributedConfiguration.ROLES.MASTER)
                .collect(Collectors.toSet())
                .size();

      } else {
        cfg = null;
        nodesConcurToTheQuorum = new HashSet<String>(iNodes);
        onlineMasters = availableNodes;
      }

      final int expectedResponses = localResult != null ? availableNodes + 1 : availableNodes;

      final int quorum =
          calculateQuorum(
              task.getQuorumType(),
              cfg,
              expectedResponses,
              nodesConcurToTheQuorum.size(),
              onlineMasters,
              checkNodesAreOnline,
              this.nodeName);

      final boolean groupByResponse =
          task.getResultStrategy() != OAbstractRemoteTask.RESULT_STRATEGY.UNION;

      final boolean waitLocalNode = waitForLocalNode(cfg, iNodes);

      // CREATE THE RESPONSE MANAGER
      final ODistributedResponseManager currentResponseMgr =
          responseManagerFactory.newResponseManager(
              iRequest,
              iNodes,
              task,
              nodesConcurToTheQuorum,
              availableNodes,
              expectedResponses,
              quorum,
              groupByResponse,
              waitLocalNode);

      if (localResult != null && currentResponseMgr.setLocalResult(this.nodeName, localResult)) {
        // COLLECT LOCAL RESULT ONLY
        return currentResponseMgr.getFinalResponse();
      }

      // SORT THE NODE TO GUARANTEE THE SAME ORDER OF DELIVERY
      if (!(iNodes instanceof List)) iNodes = new ArrayList<String>(iNodes);
      if (iNodes.size() > 1) Collections.sort((List<String>) iNodes);

      getMessageService().registerRequest(iRequest.getId().getMessageId(), currentResponseMgr);

      for (String node : iNodes) {
        // CATCH ANY EXCEPTION LOG IT AND IGNORE TO CONTINUE SENDING REQUESTS TO OTHER NODES
        try {
          final ORemoteServerController remoteServer = getRemoteServer(node);

          remoteServer.sendRequest(iRequest);

        } catch (Exception e) {
          currentResponseMgr.removeServerBecauseUnreachable(node);

          String reason = e.getMessage();
          if (e instanceof ODistributedException && e.getCause() instanceof IOException) {
            // CONNECTION ERROR: REMOVE THE CONNECTION
            reason = e.getCause().getMessage();
            closeRemoteServer(node);

          } else if (e instanceof OSecurityAccessException) {
            // THE CONNECTION COULD BE STALE, CREATE A NEW ONE AND RETRY
            closeRemoteServer(node);
            try {
              final ORemoteServerController remoteServer = getRemoteServer(node);
              remoteServer.sendRequest(iRequest);
              continue;

            } catch (Exception ex) {
              // IGNORE IT BECAUSE MANAGED BELOW
            }
          }

          if (!isNodeAvailable(node))
            // NODE IS NOT AVAILABLE
            logger.debugOut(
                this.nodeName,
                node,
                "Error on sending distributed request %s. The target node is not available. Active"
                    + " nodes: %s",
                e,
                iRequest,
                getAvailableNodeNames(databaseName));
          else
            logger.errorOut(
                this.nodeName,
                node,
                "Error on sending distributed request %s (err=%s). Active nodes: %s",
                iRequest,
                reason,
                getAvailableNodeNames(databaseName));
        }
      }

      if (currentResponseMgr.getExpectedNodes().isEmpty())
        // NO SERVER TO SEND A MESSAGE
        throw new ODistributedException(
            "No server active for distributed request ("
                + iRequest
                + ") against database '"
                + databaseName
                + "' to nodes "
                + iNodes);

      if (databaseName != null) {
        ODistributedDatabaseImpl shared = getDatabase(databaseName);
        if (shared != null) {
          shared.incSentRequest();
        }
      }

      if (iExecutionMode == ODistributedRequest.EXECUTION_MODE.RESPONSE)
        return waitForResponse(iRequest, currentResponseMgr);

      return null;

    } catch (RuntimeException e) {
      throw e;
    } catch (Exception e) {
      throw OException.wrapException(
          new ODistributedException(
              "Error on executing distributed request ("
                  + iRequest
                  + ") against database '"
                  + this.nodeName
                  + "' to nodes "
                  + iNodes),
          e);
    }
  }

  protected ODistributedResponse waitForResponse(
      final ODistributedRequest iRequest, final ODistributedResponseManager currentResponseMgr)
      throws InterruptedException {
    final long beginTime = System.currentTimeMillis();

    // WAIT FOR THE MINIMUM SYNCHRONOUS RESPONSES (QUORUM)
    if (!currentResponseMgr.waitForSynchronousResponses()) {
      final long elapsed = System.currentTimeMillis() - beginTime;

      if (elapsed > currentResponseMgr.getSynchTimeout()) {

        logger.warnIn(
            this.nodeName,
            null,
            "Timeout (%dms) on waiting for synchronous responses from nodes=%s responsesSoFar=%s"
                + " request=(%s)",
            elapsed,
            currentResponseMgr.getExpectedNodes(),
            currentResponseMgr.getRespondingNodes(),
            iRequest);
      }
    }

    return currentResponseMgr.getFinalResponse();
  }

  protected int calculateQuorum(
      final OCommandDistributedReplicateRequest.QUORUM_TYPE quorumType,
      final ODistributedConfiguration cfg,
      final int totalServers,
      final int totalMasterServers,
      int onlineMasters,
      final boolean checkNodesAreOnline,
      final String localNodeName) {

    int quorum = 1;

    int totalServerInQuorum = totalServers;
    int clusterQuorum = 0;
    switch (quorumType) {
      case NONE:
        // IGNORE IT
        break;
      case READ:
        if (cfg != null) {
          clusterQuorum = cfg.getReadQuorum(totalServers, localNodeName);
        } else {
          clusterQuorum = 1;
        }
        break;
      case WRITE:
        if (cfg != null) {
          clusterQuorum = cfg.getWriteQuorum(totalMasterServers, localNodeName);
          totalServerInQuorum = totalMasterServers;
        } else {
          clusterQuorum = totalMasterServers / 2 + 1;
          totalServerInQuorum = totalMasterServers;
        }
        break;
      case WRITE_ALL_MASTERS:
        if (cfg != null) {
          int cfgQuorum = cfg.getWriteQuorum(totalMasterServers, localNodeName);
          clusterQuorum = Math.max(cfgQuorum, onlineMasters);
        } else {
          clusterQuorum = totalMasterServers;
          totalServerInQuorum = totalMasterServers;
        }
        break;
      case ALL:
        clusterQuorum = totalServers;
        break;
    }
    quorum = Math.max(quorum, clusterQuorum);

    if (quorum < 0) quorum = 0;

    if (checkNodesAreOnline && quorum > totalServerInQuorum)
      throw new ODistributedException(
          "Quorum ("
              + quorum
              + ") cannot be reached on server '"
              + localNodeName
              + "' database '"
              + this.nodeName
              + "' because it is major than available nodes ("
              + totalServerInQuorum
              + ")");

    return quorum;
  }

  private long adjustTimeoutWithLatency(
      final Collection<String> iNodes, final long timeout, final ODistributedRequestId requestId) {
    long delta = 0;
    if (iNodes != null)
      for (String n : iNodes) {
        // UPDATE THE TIMEOUT WITH THE CURRENT SERVER LATENCY
        final long l = getMessageService().getCurrentLatency(n);
        delta = Math.max(delta, l);
      }

    return timeout + delta;
  }

  public ODistributedResponse send2Nodes(
      final ODistributedRequest iRequest,
      Collection<String> iNodes,
      final ODistributedRequest.EXECUTION_MODE iExecutionMode,
      final Object localResult) {
    return send2Nodes(
        iRequest,
        iNodes,
        iExecutionMode,
        localResult,
        (iRequest1,
            iNodes1,
            task,
            nodesConcurToTheQuorum,
            availableNodes,
            expectedResponses,
            quorum,
            groupByResponse,
            waitLocalNode) -> {
          return new ODistributedResponseManagerImpl(
              this,
              iRequest,
              iNodes,
              nodesConcurToTheQuorum,
              quorum,
              waitLocalNode,
              adjustTimeoutWithLatency(
                  iNodes, task.getSynchronousTimeout(expectedResponses), iRequest.getId()),
              groupByResponse);
        });
  }

  protected boolean waitForLocalNode(
      final ODistributedConfiguration cfg, final Collection<String> iNodes) {
    boolean waitLocalNode = false;
    if (iNodes.contains(this.nodeName)) {
      if (cfg != null) {
        if (cfg.isReadYourWrites(null)) waitLocalNode = true;
      } else {
        waitLocalNode = true;
      }
    }
    return waitLocalNode;
  }

  @Override
  public void executeOnLocalNodeFromRemote(ODistributedRequest request) {
    Object response = executeOnLocalNode(request.getId(), request.getTask(), null);
    ODistributedDatabaseImpl.sendResponseBack(this, this, request.getId(), response);
  }

  /** Executes the request on local node. In case of error returns the Exception itself */
  @Override
  public Object executeOnLocalNode(
      final ODistributedRequestId reqId,
      final ORemoteTask task,
      final ODatabaseDocumentInternal database) {

    return OScenarioThreadLocal.executeAsDistributed(
        () -> {
          return localInternalExecute(reqId, task, database);
        });
  }

  private Object localInternalExecute(
      final ODistributedRequestId reqId,
      final ORemoteTask task,
      final ODatabaseDocumentInternal database) {
    final ODistributedPlugin manager = this;
    try {
      final Object result = task.execute(reqId, serverInstance, manager, database);

      if (result instanceof Throwable && !(result instanceof OException))
        // EXCEPTION
        logger.debugIn(
            nodeName,
            getNodeNameById(reqId.getNodeId()),
            "Error on executing request %d (%s) on local node: ",
            (Throwable) result,
            reqId,
            task);

      return result;

    } catch (InterruptedException e) {
      // IGNORE IT
      logger.debugIn(
          nodeName,
          getNodeNameById(reqId.getNodeId()),
          "Interrupted execution on executing distributed request %s on local node: %s",
          e,
          reqId,
          task);
      return e;

    } catch (Exception e) {
      if (!(e instanceof OException))
        logger.errorIn(
            nodeName,
            getNodeNameById(reqId.getNodeId()),
            "Error on executing distributed request %s on local node: %s",
            e,
            reqId,
            task);

      return e;
    }
  }

  public Set<String> getManagedDatabases() {
    return getDatabases();
  }

  public String getLocalNodeName() {
    return nodeName;
  }

  @Override
  public List<String> getOnlineNodes(String iDatabaseName) {
    return clusterManager.getOnlineNodes(iDatabaseName);
  }

  @Override
  public List<String> getOnlineNodesNotLocal(String iDatabaseName) {
    List<String> list = clusterManager.getOnlineNodes(iDatabaseName);
    list.remove(getLocalNodeName());
    return list;
  }

  @Override
  public void reassignClustersOwnership(
      final String iNode, final String databaseName, final boolean canCreateNewClusters) {
    final ODistributedConfiguration cfg = getDatabaseConfiguration(databaseName);
    final ODistributedConfiguration.ROLES role = cfg.getServerRole(iNode);
    if (role != ODistributedConfiguration.ROLES.MASTER)
      // NO MASTER, DON'T CREATE LOCAL CLUSTERS
      return;

    ODatabaseDocumentInternal current = ODatabaseRecordThreadLocal.instance().getIfDefined();
    try (ODatabaseDocumentInternal iDatabase = getServerInstance().openDatabase(databaseName)) {

      logger.infoNode(
          nodeName, "Reassigning ownership of clusters for database %s...", iDatabase.getName());

      final Set<String> availableNodes = getAvailableNodeNames(iDatabase.getName());

      // FILTER OUT NON MASTER SERVER
      for (Iterator<String> it = availableNodes.iterator(); it.hasNext(); ) {
        final String node = it.next();
        if (cfg.getServerRole(node) != ODistributedConfiguration.ROLES.MASTER) it.remove();
      }
      iDatabase.activateOnCurrentThread();
      final OSchema schema = iDatabase.getDatabaseOwner().getMetadata().getSchema();

      for (final OClass clazz : schema.getClasses()) {
        ((OClassDistributed) clazz)
            .autoAssignClusterOwnership(iDatabase, availableNodes, canCreateNewClusters);
      }

      logger.infoNode(
          nodeName,
          "Reassignment of clusters for database '%s' completed (classes=%d)",
          iDatabase.getName(),
          schema.getClasses().size());
    } finally {
      ODatabaseRecordThreadLocal.instance().set(current);
    }
  }

  @Override
  public boolean isNodeAvailable(String iNodeName, String databaseName) {
    return clusterManager.isNodeAvailable(iNodeName, databaseName);
  }

  @Override
  public boolean isNodeOnline(String iNodeName, String databaseName) {
    return clusterManager.isNodeOnline(iNodeName, databaseName);
  }

  @Override
  public boolean isNodeStatusEqualsTo(
      final String iNodeName, final String iDatabaseName, final DB_STATUS... statuses) {
    final DB_STATUS s = getDatabaseStatus(iNodeName, iDatabaseName);
    for (DB_STATUS st : statuses) {
      if (s == st) return true;
    }
    return false;
  }

  @Override
  public boolean isNodeAvailable(String iNodeName) {
    return clusterManager.isNodeAvailable(iNodeName);
  }

  @Override
  public Set<String> getAvailableNodeNames(String databaseName) {
    return clusterManager.getAvailableNodeNames(databaseName);
  }

  @Override
  public Set<String> getAvailableNodeNotLocalNames(String databaseName) {
    Set<String> set = clusterManager.getAvailableNodeNames(databaseName);
    set.remove(getLocalNodeName());
    return set;
  }

  public boolean isOffline() {
    return getNodeStatus() != NODE_STATUS.ONLINE;
  }

  @Override
  public int getLocalNodeId() {
    return clusterManager.getLocalNodeId();
  }

  /** Returns the nodes with the requested status. */
  @Override
  public int getNodesWithStatus(
      final Collection<String> iNodes, final String databaseName, final DB_STATUS... statuses) {
    for (Iterator<String> it = iNodes.iterator(); it.hasNext(); ) {
      final String node = it.next();

      if (!isNodeStatusEqualsTo(node, databaseName, statuses)) it.remove();
    }
    return iNodes.size();
  }

  @Override
  public String toString() {
    return nodeName;
  }

  @Override
  public ODistributedMessageService getMessageService() {
    return ((OrientDBDistributed) serverInstance.getDatabases()).getMessageService();
  }

  @Override
  public int getAvailableNodes(String iDatabaseName) {
    return clusterManager.getAvailableNodes(iDatabaseName);
  }

  @Override
  public boolean installDatabase(
      final boolean iStartup,
      final String databaseName,
      final boolean forceDeployment,
      final boolean tryWithDeltaFirst) {
    if (getDatabaseStatus(getLocalNodeName(), databaseName) == DB_STATUS.OFFLINE)
      // OFFLINE: AVOID TO INSTALL IT
      return false;

    if (databaseName.equalsIgnoreCase(OSystemDatabase.SYSTEM_DB_NAME))
      // DON'T REPLICATE SYSTEM BECAUSE IS DIFFERENT AND PER SERVER
      return false;

    if (!installingDatabases.add(databaseName)) {
      return false;
    }

    try {
      return executeInDistributedDatabaseLock(
          databaseName,
          20000,
          () -> {
            return internalInstallDatabase(
                iStartup, databaseName, forceDeployment, tryWithDeltaFirst);
          });
    } finally {
      installingDatabases.remove(databaseName);
    }
  }

  public boolean isSyncronizing(String databaseName) {
    return this.installingDatabases.contains(databaseName);
  }

  public Boolean internalInstallDatabase(
      final boolean iStartup,
      final String databaseName,
      final boolean forceDeployment,
      final boolean tryWithDeltaFirst) {
    OrientDBDistributed context = (OrientDBDistributed) getServerInstance().getDatabases();
    ODistributedConfiguration cfg = context.getExistingDistributedConfiguration(databaseName);
    if (cfg == null) {
      cfg = context.getDefaultDistributedConfiguration(databaseName);
    }

    // GET ALL THE OTHER SERVERS
    final Collection<String> nodes = nodesOnlineNotSelf(databaseName);
    if (nodes.size() == 0) {
      logger.errorNode(
          nodeName,
          "Cannot install database '%s' on local node, because no servers are available",
          databaseName);
      return false;
    }

    logger.infoNode(
        nodeName,
        "Current node is a %s for database '%s'",
        cfg.getServerRole(nodeName),
        databaseName);

    if (!forceDeployment && getDatabaseStatus(getLocalNodeName(), databaseName) == DB_STATUS.ONLINE)
      return false;

    context.distributedPauseDatabase(databaseName);

    final Boolean deploy = forceDeployment ? Boolean.TRUE : (Boolean) cfg.isAutoDeploy();

    boolean databaseInstalled;

    try {

      // CREATE THE DISTRIBUTED QUEUE
      // TODO: This should check also but can't do it now
      // storage.getLastMetadata().isPresent();
      if (!context.exists(databaseName, null, null)) {

        if (deploy == null || !deploy) {
          context.distributedSetOnline(databaseName);
          return false;
        }

        // FIRST TIME, ASK FOR FULL REPLICA
        databaseInstalled = requestFullDatabase(databaseName, iStartup);

      } else {
        if (tryWithDeltaFirst) {
          try {

            // TRY WITH DELTA SYNC
            databaseInstalled = requestNewDatabaseDelta(databaseName);

          } catch (ODistributedDatabaseDeltaSyncException e) {
            if (deploy == null || !deploy) {
              context.distributedSetOnline(databaseName);
              return false;
            }

            databaseInstalled = requestFullDatabase(databaseName, iStartup);
          }
        } else
          // SKIP DELTA AND EXECUTE FULL BACKUP
          databaseInstalled = requestFullDatabase(databaseName, iStartup);
      }

      if (!databaseInstalled) {
        setDatabaseStatus(getLocalNodeName(), databaseName, DB_STATUS.NOT_AVAILABLE);
      }

    } catch (ODatabaseIsOldException e) {

      // CURRENT DATABASE IS NEWER, SET ALL OTHER DATABASES AS NOT_AVAILABLE TO FORCE THEM
      // TO ASK FOR THE CURRENT DATABASE
      context.distributedSetOnline(databaseName);

      logger.infoOut(
          nodeName,
          null,
          "Current copy of database '%s' is newer than the copy present in the cluster. Use the"
              + " local copy and force other nodes to download this",
          databaseName);

      databaseInstalled = true;
    } catch (RuntimeException e) {
      // UNLOCK ACCEPTING REQUESTS EVEN IN CASE OF ERROR.
      context.distributedSetOnline(databaseName);
      throw e;
    }

    return databaseInstalled;
  }

  public void checkNodeInConfiguration(final String databaseName, ODistributedConfiguration cfg) {
    executeInDistributedDatabaseLock(
        databaseName,
        20000,
        cfg != null ? cfg.modify() : null,
        lastCfg -> {
          return internalCheckNodeInConfig(databaseName, lastCfg);
        });
  }

  public Object internalCheckNodeInConfig(
      final String databaseName, OModifiableDistributedConfiguration lastCfg) {
    // GET LAST VERSION IN LOCK
    final List<String> foundPartition = lastCfg.addNewNodeInServerList(nodeName);
    if (foundPartition != null) {
      logger.infoNode(
          nodeName,
          "Adding node '%s' in partition: %s db=%s v=%d",
          nodeName,
          foundPartition,
          databaseName,
          lastCfg.getVersion());
    }
    return null;
  }

  private Collection<String> nodesOnlineNotSelf(String db) {
    Collection<String> nodes = new HashSet<String>(getActiveServers());
    if (nodes.isEmpty()) {
      return nodes;
    }
    nodes.remove(getLocalNodeName());
    final List<String> selectedNodes = new ArrayList<String>();

    for (String n : nodes) {
      if (isNodeStatusEqualsTo(n, db, DB_STATUS.ONLINE, DB_STATUS.BACKUP)) {
        selectedNodes.add(n);
        break;
      }
    }
    if (selectedNodes.isEmpty()) {
      // NO NODE ONLINE, SEND THE MESSAGE TO EVERYONE
      for (String n : nodes) {
        if (isNodeAvailable(n)) {
          selectedNodes.add(n);
        }
      }
    }
    return selectedNodes;
  }

  private boolean requestNewDatabaseDelta(String databaseName) {
    // GET ALL THE OTHER SERVERS
    final Collection<String> nodes = nodesOnlineNotSelf(databaseName);
    if (nodes.size() == 0) {
      return false;
    }

    logger.warnOut(
        nodeName,
        nodes.toString(),
        "requesting delta database sync for '%s' on local server...",
        databaseName);

    boolean databaseInstalledCorrectly = false;

    for (String targetNode : nodes) {

      if (!isNodeOnline(targetNode, databaseName)) {
        continue;
      }
      OTxMetadataHolder metadata;
      try (ODatabaseDocumentInternal inst =
          serverInstance.getDatabases().openNoAuthorization(databaseName)) {
        Optional<byte[]> read = inst.getStorage().getLastMetadata();
        if (read.isPresent()) {
          metadata = OTxMetadataHolderImpl.read(read.get());
        } else {
          throw new ODistributedDatabaseDeltaSyncException("Trigger full sync");
        }
      }
      final OSyncDatabaseNewDeltaTask deployTask =
          new OSyncDatabaseNewDeltaTask(metadata.getStatus());

      try {
        final ODistributedResponse response =
            sendSingleRequest(databaseName, targetNode, deployTask);

        if (response == null)
          throw new ODistributedDatabaseDeltaSyncException("Error requesting delta sync");

        databaseInstalledCorrectly =
            installResponseNewDeltaSync(
                databaseName, targetNode, (ONewDeltaTaskResponse) response.getPayload());

      } catch (ODistributedDatabaseDeltaSyncException e) {
        // RE-THROW IT
        throw e;
      } catch (Exception e) {
        logger.errorOut(
            nodeName,
            targetNode,
            "Error on asking delta backup of database '%s' (err=%s)",
            databaseName,
            e.getMessage());
        throw OException.wrapException(new ODistributedDatabaseDeltaSyncException(e.toString()), e);
      }

      if (databaseInstalledCorrectly) {
        try {
          reassignClustersOwnership(nodeName, databaseName, true);
        } catch (Exception e) {
          // HANDLE IT AS WARNING
          logger.warnNode(
              nodeName, "Error on re-balancing the cluster for database '%s'", e, databaseName);
          // NOT CRITICAL, CONTINUE
        }
        return true;
      }
    }

    throw new ODistributedDatabaseDeltaSyncException("Requested database delta sync error");
  }

  protected boolean requestFullDatabase(final String databaseName, final boolean backupDatabase) {
    logger.infoNode(nodeName, "Requesting full sync for database '%s'...", databaseName);

    for (int retry = 0; retry < DEPLOY_DB_MAX_RETRIES; ++retry) {
      // ASK DATABASE TO THE FIRST NODE, THE FIRST ATTEMPT, OTHERWISE ASK TO EVERYONE
      if (requestDatabaseFullSync(backupDatabase, databaseName))
        // DEPLOYED
        return true;
      try {
        Thread.sleep(
            serverInstance.getContextConfiguration().getValueAsLong(DISTRIBUTED_MAX_STARTUP_DELAY));
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        return false;
      }
    }
    // RETRY COUNTER EXCEED
    return false;
  }

  private boolean installResponseNewDeltaSync(
      String databaseName, String targetNode, ONewDeltaTaskResponse results) {
    final String dbPath = serverInstance.getDatabaseDirectory() + databaseName;
    boolean databaseInstalledCorrectly = false;
    // EXTRACT THE REAL RESULT
    if (results.getResponseType() == ONewDeltaTaskResponse.ResponseType.CHUNK) {
      ODistributedDatabaseChunk firstChunk = results.getChunk().get();
      try {

        OSyncReceiver receiver =
            new OSyncReceiver(this, databaseName, firstChunk, targetNode, dbPath);
        receiver.spawnReceiverThread();
        receiver.getStarted().await();

        try (InputStream in = receiver.getInputStream()) {
          OrientDBInternal context = serverInstance.getDatabases();
          databaseInstalledCorrectly =
              context.deltaSync(databaseName, in, OrientDBConfig.defaultConfig());
        } catch (IOException e) {
          throw OException.wrapException(
              new OIOException("Error on distributed sync of database"), e);
        }
        if (databaseInstalledCorrectly) {
          logger.infoIn(nodeName, targetNode, "Installed delta of database '%s'", databaseName);
        }

        // DATABASE INSTALLED CORRECTLY

      } catch (OException | InterruptedException e) {
        logger.error("Error installing database from network", e);
        databaseInstalledCorrectly = false;
      }
    } else if (results.getResponseType() == ONewDeltaTaskResponse.ResponseType.FULL_SYNC) {
      throw new ODistributedDatabaseDeltaSyncException("Full sync required");
    } else if (results.getResponseType() == ONewDeltaTaskResponse.ResponseType.NO_CHANGES) {
      OrientDBInternal context = serverInstance.getDatabases();
      context.distributedSetOnline(databaseName);
      return true;
    }
    return databaseInstalledCorrectly;
  }

  protected boolean requestDatabaseFullSync(
      final boolean backupDatabase, final String databaseName) {
    final Collection<String> nodes = nodesOnlineNotSelf(databaseName);
    if (nodes.isEmpty()) {
      logger.warnNode(
          nodeName,
          "Cannot request full deploy of database '%s' because there are no nodes available with"
              + " such database",
          databaseName);
      return false;
    }

    logger.infoOut(
        nodeName,
        nodes.toString(),
        "Requesting deploy of database '%s' on local server...",
        databaseName);
    for (String nodetoSend : nodes) {
      OSyncDatabaseTask deployTask = new OSyncDatabaseTask();
      ODistributedResponse response = sendSingleRequest(databaseName, nodetoSend, deployTask);

      if (response == null || response.getPayload() == null) {
        logger.errorIn(
            nodeName,
            nodes.toString(),
            "Timeout waiting the sync database please set the `distributed.deployDbTaskTimeout` to"
                + " appropriate value");
        setDatabaseStatus(nodeName, databaseName, DB_STATUS.NOT_AVAILABLE);
        return false;
      }

      final Object value = response.getPayload();
      final String dbPath = serverInstance.getDatabaseDirectory() + databaseName;

      if (value instanceof ODistributedDatabaseChunk) {
        if (backupDatabase) backupCurrentDatabase(databaseName);

        try {
          installDatabaseFromNetwork(
              dbPath, databaseName, nodetoSend, (ODistributedDatabaseChunk) value);
        } catch (OException e) {
          logger.error("Error installing database from network", e);
          continue;
        }

        return true;
      } else if (value instanceof Boolean) {
        serverInstance.getDatabases().distributedSetOnline(databaseName);
        continue;
      } else if (value instanceof ODatabaseIsOldException) {

        // MANAGE THIS EXCEPTION AT UPPER LEVEL
        throw (ODatabaseIsOldException) value;

      } else if (value instanceof Throwable) {
        logger.errorIn(
            nodeName,
            nodetoSend,
            "Error on installing database '%s' in %s",
            (Throwable) value,
            databaseName,
            dbPath);

        setDatabaseStatus(nodeName, databaseName, DB_STATUS.NOT_AVAILABLE);

        if (value instanceof ODistributedException) throw (ODistributedException) value;

      } else throw new IllegalArgumentException("Type " + value + " not supported");
    }

    throw new ODistributedException(
        "No response received from remote nodes for auto-deploy of database '"
            + databaseName
            + "'");
  }

  protected void backupCurrentDatabase(final String iDatabaseName) {
    serverInstance.getDatabases().forceDatabaseClose(iDatabaseName);

    final String backupDirectory =
        serverInstance
            .getContextConfiguration()
            .getValueAsString(OGlobalConfiguration.DISTRIBUTED_BACKUP_DIRECTORY);

    Path serverDir = Paths.get(serverInstance.getDatabaseDirectory());
    Path backupPath = serverDir.resolve(backupDirectory).toAbsolutePath();

    backupPath = backupPath.resolve(iDatabaseName);

    final String dbpath = serverInstance.getDatabaseDirectory() + iDatabaseName;
    final File backupFullPath = backupPath.toFile();
    try {
      if (backupFullPath.exists()) {
        OFileUtils.deleteRecursively(backupFullPath);
      }

      Files.createDirectories(backupFullPath.toPath());

      // move the database on current node
      logger.warnNode(
          nodeName,
          "Moving existent database '%s' in '%s' to '%s' and get a fresh copy from a remote"
              + " node...",
          iDatabaseName,
          dbpath,
          backupPath);

      final File oldDirectory = new File(dbpath);
      if (oldDirectory.exists() && oldDirectory.isDirectory()) {
        if (oldDirectory.getCanonicalPath().equals(backupFullPath.getCanonicalPath())) {
          throw new ODistributedException(
              String.format(
                  "Backup folder configured as same of database folder:'%s'",
                  oldDirectory.getAbsolutePath()));
        }
        try {
          try {
            Files.move(
                oldDirectory.toPath(), backupFullPath.toPath(), StandardCopyOption.ATOMIC_MOVE);
          } catch (AtomicMoveNotSupportedException e) {
            logger.errorNoDb(
                "Atomic moves not supported during database backup, will try not atomic move",
                null);
            if (backupFullPath.exists()) {
              OFileUtils.deleteRecursively(backupFullPath);
            }
            Files.createDirectories(backupFullPath.toPath());

            Files.move(oldDirectory.toPath(), backupPath.resolve(oldDirectory.getName()));
          }
        } catch (DirectoryNotEmptyException e) {
          logger.errorNoDb(
              "File rename not supported during database backup, will try coping files", null);
          if (backupFullPath.exists()) {
            OFileUtils.deleteRecursively(backupFullPath);
          }
          Files.createDirectories(backupFullPath.toPath());
          try {
            OFileUtils.copyDirectory(
                oldDirectory, backupPath.resolve(oldDirectory.getName()).toFile());
            OFileUtils.deleteRecursively(backupFullPath);
          } catch (IOException ioe) {
            logger.errorNoDb("Error moving old database removing it", ioe);
            OFileUtils.deleteRecursively(backupFullPath);
          }
        }
      }
    } catch (IOException e) {
      logger.warnNode(
          nodeName,
          "Error on moving existent database '%s' located in '%s' to '%s' (error=%s).",
          e,
          iDatabaseName,
          dbpath,
          backupFullPath,
          e);
    }
  }

  /** Installs a database from the network. */
  protected void installDatabaseFromNetwork(
      final String dbPath,
      final String databaseName,
      final String iNode,
      final ODistributedDatabaseChunk firstChunk) {

    OSyncReceiver receiver = new OSyncReceiver(this, databaseName, firstChunk, iNode, dbPath);
    receiver.spawnReceiverThread();

    installDatabaseOnLocalNode(databaseName, dbPath, iNode, firstChunk.incremental, receiver);
    receiver.close();

    try {
      reassignClustersOwnership(nodeName, databaseName, false);
    } catch (Exception e) {
      // HANDLE IT AS WARNING
      logger.warnNode(
          nodeName, "Error on re-balancing the cluster for database '%s'", e, databaseName);
      // NOT CRITICAL, CONTINUE
    }
  }

  @Override
  public ORemoteTaskFactoryManager getTaskFactoryManager() {
    return taskFactoryManager;
  }

  @Override
  public Set<String> getActiveServers() {
    return clusterManager.getActiveServers();
  }

  @Override
  public Set<String> getActiveServerNotLocal() {
    Set<String> res = clusterManager.getActiveServers();
    res.remove(getLocalNodeName());
    return res;
  }

  public ODistributedStrategy getDistributedStrategy() {
    return responseManagerFactory;
  }

  public void setDistributedStrategy(final ODistributedStrategy streatgy) {
    this.responseManagerFactory = streatgy;
  }

  @Override
  public boolean updateCachedDatabaseConfiguration(
      String iDatabaseName, OModifiableDistributedConfiguration cfg) {
    return clusterManager.updateCachedDatabaseConfiguration(iDatabaseName, cfg);
  }

  @Override
  public void publishDistributedConfiguration(String iDatabaseName, ODistributedConfiguration cfg) {
    clusterManager.publishDistributedConfiguration(iDatabaseName, cfg);
  }

  public void notifyClients(String databaseName) {
    List<String> hosts = new ArrayList<>();
    for (String name : getActiveServers()) {
      ONodeConfig memberConfig = clusterManager.getNodeConfigurationByName(name, true);
      if (memberConfig != null) {
        final String nodeStatus = memberConfig.getStatus();

        if (!"OFFLINE".equals(nodeStatus)) {
          final Collection<ONodeListenerConfig> listeners = memberConfig.getListeners();
          if (listeners != null)
            for (ONodeListenerConfig listener : listeners) {
              if (listener.getProtocol().equals("ONetworkProtocolBinary")) {
                hosts.add(listener.getListen());
              }
            }
        }
      }
    }
    serverInstance.getPushManager().pushDistributedConfig(databaseName, hosts);
  }

  public void onDatabaseEvent(
      final String nodeName, final String databaseName, final DB_STATUS status) {
    notifyClients(databaseName);
    updateLastClusterChange();
    dumpServersStatus();
  }

  public void invokeOnDatabaseStatusChange(
      final String iNode, final String iDatabaseName, final DB_STATUS iStatus) {
    // NOTIFY DB/NODE IS CHANGING STATUS
    for (ODistributedLifecycleListener l : listeners) {
      try {
        l.onDatabaseChangeStatus(iNode, iDatabaseName, iStatus);
      } catch (Exception e) {
        // IGNORE IT
      }
    }
  }

  protected void assignNodeName() {
    // ORIENTDB_NODE_NAME ENV VARIABLE OR JVM SETTING
    nodeName = OSystemVariableResolver.resolveVariable(NODE_NAME_ENV);

    if (nodeName != null) {
      nodeName = nodeName.trim();
      if (nodeName.isEmpty()) nodeName = null;
    }

    if (nodeName == null) {
      try {
        // WAIT ANY LOG IS PRINTED
        Thread.sleep(1000);
      } catch (InterruptedException e) {
      }

      System.out.println();
      System.out.println();
      System.out.println(
          OAnsiCode.format(
              "$ANSI{yellow +---------------------------------------------------------------+}"));
      System.out.println(
          OAnsiCode.format(
              "$ANSI{yellow |         WARNING: FIRST DISTRIBUTED RUN CONFIGURATION          |}"));
      System.out.println(
          OAnsiCode.format(
              "$ANSI{yellow +---------------------------------------------------------------+}"));
      System.out.println(
          OAnsiCode.format(
              "$ANSI{yellow | This is the first time that the server is running as          |}"));
      System.out.println(
          OAnsiCode.format(
              "$ANSI{yellow | distributed. Please type the name you want to assign to the   |}"));
      System.out.println(
          OAnsiCode.format(
              "$ANSI{yellow | current server node.                                          |}"));
      System.out.println(
          OAnsiCode.format(
              "$ANSI{yellow |                                                               |}"));
      System.out.println(
          OAnsiCode.format(
              "$ANSI{yellow | To avoid this message set the environment variable or JVM     |}"));
      System.out.println(
          OAnsiCode.format(
              "$ANSI{yellow | setting ORIENTDB_NODE_NAME to the server node name to use.    |}"));
      System.out.println(
          OAnsiCode.format(
              "$ANSI{yellow +---------------------------------------------------------------+}"));
      System.out.print(OAnsiCode.format("\n$ANSI{yellow Node name [BLANK=auto generate it]: }"));

      OConsoleReader reader = new ODefaultConsoleReader();
      try {
        nodeName = reader.readLine();
      } catch (IOException e) {
      }
      if (nodeName != null) {
        nodeName = nodeName.trim();
        if (nodeName.isEmpty()) nodeName = null;
      }
    }

    if (nodeName == null)
      // GENERATE NODE NAME
      this.nodeName = "node" + System.currentTimeMillis();

    logger.warnNode("Assigning distributed node name: %s", this.nodeName);

    // SALVE THE NODE NAME IN CONFIGURATION
    boolean found = false;
    final OServerConfiguration cfg = serverInstance.getConfiguration();
    for (OServerHandlerConfiguration h : cfg.handlers) {
      if (h.clazz.equals(getClass().getName())) {
        for (OServerParameterConfiguration p : h.parameters) {
          if (p.name.equals("nodeName")) {
            found = true;
            p.value = this.nodeName;
            break;
          }
        }

        if (!found) {
          h.parameters = OArrays.copyOf(h.parameters, h.parameters.length + 1);
          h.parameters[h.parameters.length - 1] =
              new OServerParameterConfiguration("nodeName", this.nodeName);
        }

        try {
          serverInstance.saveConfiguration();
        } catch (IOException e) {
          throw OException.wrapException(
              new OConfigurationException("Cannot save server configuration"), e);
        }
        break;
      }
    }
  }

  protected void installDatabaseOnLocalNode(
      final String databaseName,
      final String dbPath,
      final String iNode,
      boolean incremental,
      OSyncReceiver receiver) {
    logger.infoIn(nodeName, iNode, "Installing database '%s' to: %s...", databaseName, dbPath);

    new File(dbPath).mkdirs();
    try {
      receiver.getStarted().await();
    } catch (InterruptedException e) {
      throw OException.wrapException(
          new OInterruptedException("Interrupted waiting receive of sync"), e);
    }

    executeInDistributedDatabaseLock(
        databaseName,
        20000,
        () -> {
          return internalInstallDatabase(databaseName, iNode, incremental, receiver);
        });
  }

  private Object internalInstallDatabase(
      final String database, final String node, boolean incremental, OSyncReceiver receiver) {
    try {
      OrientDBDistributed context = (OrientDBDistributed) serverInstance.getDatabases();
      if (incremental) {
        context.fullSync(database, receiver.getInputStream(), OrientDBConfig.defaultConfig());
        context.saveDatabaseConfiguration(database);

        try (ODatabaseDocumentInternal inst = context.openNoAuthorization(database)) {
          Optional<byte[]> read = inst.getStorage().getLastMetadata();
          if (read.isPresent()) {
            OTxMetadataHolder metadata = OTxMetadataHolderImpl.read(read.get());
            final OSyncDatabaseNewDeltaTask deployTask =
                new OSyncDatabaseNewDeltaTask(metadata.getStatus());

            final ODistributedResponse response = sendSingleRequest(database, node, deployTask);
            if (response == null)
              throw new ODistributedDatabaseDeltaSyncException("Error Requesting delta sync");
            boolean installed =
                installResponseNewDeltaSync(
                    database, node, (ONewDeltaTaskResponse) response.getPayload());
            if (!installed)
              throw new ODistributedDatabaseDeltaSyncException("Error Requesting delta sync");
          }
        }
      } else {

        // USES A CUSTOM WRAPPER OF IS TO WAIT FOR FILE IS WRITTEN (ASYNCH)
        try (InputStream in = receiver.getInputStream()) {

          // IMPORT FULL DATABASE (LISTENER ONLY FOR DEBUG PURPOSE)
          context.networkRestore(database, in, null);
        }
      }
      return null;
    } catch (IOException e) {
      throw OException.wrapException(new OIOException("Error on distributed sync of database"), e);
    }
  }

  public ODistributedRequestId nextRequestId() {
    return new ODistributedRequestId(getLocalNodeId(), getNextMessageIdCounter());
  }

  public void stopNode(final String iNode) throws IOException {
    logger.warnNode(nodeName, "Sending request of stopping node '%s'...", iNode);

    final ODistributedRequest request =
        new ODistributedRequest(
            this,
            nextRequestId(),
            null,
            getTaskFactoryManager()
                .getFactoryByServerName(iNode)
                .createTask(OStopServerTask.FACTORYID));

    getRemoteServer(iNode).sendRequest(request);
  }

  public void restartNode(final String iNode) throws IOException {
    logger.warnNode(nodeName, "Sending request of restarting node '%s'...", iNode);

    final ODistributedRequest request =
        new ODistributedRequest(
            this,
            nextRequestId(),
            null,
            getTaskFactoryManager()
                .getFactoryByServerName(iNode)
                .createTask(ORestartServerTask.FACTORYID));

    getRemoteServer(iNode).sendRequest(request);
  }

  public long getNextMessageIdCounter() {
    return localMessageIdCounter.getAndIncrement();
  }

  @Override
  public void updateLastClusterChange() {
    clusterManager.updateLastClusterChange();
  }

  public void closeRemoteServer(final String node) {
    remoteServerManager.closeRemoteServer(node);
  }

  /** Avoids to dump the same configuration twice if it's unchanged since the last time. */
  public void dumpServersStatus() {
    final OClusterConfiguration cfg = getClusterConfiguration();

    final String compactStatus = ODistributedOutput.getCompactServerStatus(this, cfg);

    if (!lastServerDump.equals(compactStatus)) {
      lastServerDump = compactStatus;

      logger.infoNode(
          getLocalNodeName(),
          "Distributed servers status (*=current):\n%s",
          ODistributedOutput.formatServerStatus(this, cfg));
    }
  }

  public static String getListeningBinaryAddress(final ONodeConfig cfg) {
    if (cfg == null) return null;

    final Collection<ONodeListenerConfig> listeners = cfg.getListeners();
    if (listeners == null)
      throw new ODatabaseException(
          "Cannot connect to a remote node because bad distributed configuration: missing"
              + " 'listeners' array field");
    String listenUrl = null;
    for (ONodeListenerConfig listener : listeners) {
      if ((listener.getProtocol()).equals("ONetworkProtocolBinary")) {
        listenUrl = (String) listener.getListen();
        break;
      }
    }
    return listenUrl;
  }

  @Override
  public void messageReceived(ODistributedRequest request) {

    for (ODistributedLifecycleListener listener : listeners) {
      listener.onMessageReceived(request);
    }
  }

  @Override
  public void messageBeforeOp(String op, ODistributedRequestId request) {

    for (ODistributedLifecycleListener listener : listeners) {
      listener.onMessageBeforeOp(op, request);
    }
  }

  @Override
  public void messageAfterOp(String op, ODistributedRequestId request) {
    for (ODistributedLifecycleListener listener : listeners) {
      listener.onMessageAfterOp(op, request);
    }
  }

  @Override
  public void messageCurrentPayload(ODistributedRequestId requestId, Object responsePayload) {
    for (ODistributedLifecycleListener listener : listeners) {
      listener.onMessageCurrentPayload(requestId, responsePayload);
    }
  }

  @Override
  public void messageProcessStart(ODistributedRequest message) {
    for (ODistributedLifecycleListener listener : listeners) {
      listener.onMessageProcessStart(message);
    }
  }

  @Override
  public void messageProcessEnd(ODistributedRequest iRequest, Object responsePayload) {
    for (ODistributedLifecycleListener listener : listeners) {
      listener.onMessageProcessEnd(iRequest, responsePayload);
    }
  }

  /** Initializes all the available server's databases as distributed. */
  public void loadLocalDatabases() {
    ((OrientDBDistributed) serverInstance.getDatabases()).loadAllDatabases();
  }

  public void installNewDatabasesFromCluster() {
    if (getActiveServers().size() <= 1) {
      // NO OTHER NODES WHERE ALIGN
      return;
    }

    final List<String> dbs = new ArrayList<>(clusterManager.getDatabases());
    Collections.sort(dbs);

    for (String databaseName : dbs) {
      final Set<String> availableServers = getAvailableNodeNames(databaseName);
      if (availableServers.isEmpty())
        // NO NODE HAS THIS DATABASE AVAILABLE
        continue;

      final DB_STATUS currStatus = getDatabaseStatus(nodeName, databaseName);
      if (currStatus == DB_STATUS.SYNCHRONIZING
          || currStatus == DB_STATUS.ONLINE
          || currStatus == DB_STATUS.BACKUP)
        // FIX PREVIOUS STATUS OF DATABASE
        setDatabaseStatus(nodeName, databaseName, DB_STATUS.NOT_AVAILABLE);

      try {
        if (!installDatabase(true, databaseName, false, true)) {
          setDatabaseStatus(getLocalNodeName(), databaseName, DB_STATUS.NOT_AVAILABLE);
        }
      } catch (Exception e) {
        logger.errorNode(
            getLocalNodeName(),
            "Error on installing database '%s' on local node (error=%s)",
            e,
            databaseName,
            e.toString());
      }
    }
  }

  public void notifyStarted() {
    serverStarted.countDown();
  }

  protected void dumpStats() {
    try {
      final OClusterConfiguration clusterCfg = getClusterConfiguration();

      final Set<String> dbs = getManagedDatabases();

      final StringBuilder buffer = new StringBuilder(8192);

      buffer.append(ODistributedOutput.formatLatency(this, clusterCfg));
      buffer.append(ODistributedOutput.formatMessages(this, clusterCfg));

      OLogManager.instance().flush();
      for (String db : dbs) {
        buffer.append(getDatabase(db).dump());
      }

      // DUMP HA STATS
      System.out.println(buffer);

    } catch (Exception e) {
      logger.errorNode(nodeName, "Error on printing HA stats", e);
    }
  }

  public ORemoteServerController getRemoteServer(final String rNodeName) throws IOException {
    if (rNodeName == null) throw new IllegalArgumentException("Server name is NULL");

    ORemoteServerController remoteServer = remoteServerManager.getRemoteServer(rNodeName);
    if (remoteServer == null) {
      Member member = clusterManager.getClusterMemberByName(rNodeName);

      for (int retry = 0; retry < 20; ++retry) {
        ONodeConfig cfg = getNodeConfigurationByUuid(member.getUuid(), false);
        if (cfg == null || cfg.getListeners() == null) {
          try {
            Thread.sleep(100);
            member = clusterManager.getClusterMemberByName(rNodeName);
            continue;

          } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw OException.wrapException(
                new ODistributedException("Cannot find node '" + rNodeName + "'"), e);
          }
        }

        final String url = ODistributedPlugin.getListeningBinaryAddress(cfg);

        if (url == null) {
          closeRemoteServer(rNodeName);
          throw new ODatabaseException(
              "Cannot connect to a remote node because the url was not found");
        }

        final String userPassword = cfg.getReplicator();

        if (userPassword != null) {
          remoteServer =
              remoteServerManager.connectRemoteServer(
                  rNodeName, url, REPLICATOR_USER, userPassword);
          break;
        }

        // RETRY TO GET USR+PASSWORD IN A WHILE
        try {
          Thread.sleep(100);
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          throw OException.wrapException(
              new OInterruptedException("Cannot connect to remote server " + rNodeName), e);
        }
      }
    }

    if (remoteServer == null)
      throw new ODistributedException("Cannot find node '" + rNodeName + "'");

    return remoteServer;
  }

  @Override
  public long getLastClusterChangeOn() {
    return clusterManager.getLastClusterChangeOn();
  }

  @Override
  public NODE_STATUS getNodeStatus() {
    return clusterManager.getNodeStatus();
  }

  @Override
  public void setNodeStatus(NODE_STATUS iStatus) {
    clusterManager.setNodeStatus(iStatus);
  }

  public void onNodeJoined(String joinedNodeName, Member member) {
    try {
      getRemoteServer(joinedNodeName);
    } catch (IOException e) {
      logger.errorOut(nodeName, joinedNodeName, "Error on connecting to node %s", joinedNodeName);
    }

    logger.infoIn(
        nodeName,
        clusterManager.getNodeName(member, true),
        "Added node configuration id=%s name=%s, now %d nodes are configured",
        member,
        clusterManager.getNodeName(member, true),
        getActiveServers().size());

    // NOTIFY NODE WAS ADDED SUCCESSFULLY
    for (ODistributedLifecycleListener l : listeners) l.onNodeJoined(joinedNodeName);

    // FORCE THE ALIGNMENT FOR ALL THE ONLINE DATABASES AFTER THE JOIN ONLY IF AUTO-DEPLOY IS SET
    for (String db : getDatabases()) {
      if (getDatabaseConfiguration(db).isAutoDeploy()
          && getDatabaseStatus(joinedNodeName, db) == DB_STATUS.ONLINE) {
        setDatabaseStatus(joinedNodeName, db, DB_STATUS.NOT_AVAILABLE);
      }
    }
    dumpServersStatus();
  }

  // This is used only during startup and gets called by the cluster metadata manager
  public void connectToAllNodes(Set<String> clusterNodes) throws IOException {
    for (String m : clusterNodes) if (!m.equals(nodeName)) getRemoteServer(m);
  }

  @Override
  public void removeServer(final String nodeLeftName, final boolean removeOnlyDynamicServers) {
    if (nodeLeftName == null) return;
    Member member = clusterManager.removeFromLocalActiveServerList(nodeLeftName);
    if (member == null) return;

    try {
      // REMOVE INTRA SERVER CONNECTION
      closeRemoteServer(nodeLeftName);

      // NOTIFY ABOUT THE NODE HAS LEFT
      for (ODistributedLifecycleListener l : listeners)
        try {
          l.onNodeLeft(nodeLeftName);
        } catch (Exception e) {
          // IGNORE IT
          logger.debugIn(nodeName, nodeLeftName, "Error on calling onNodeLeft event on '%s'", e, l);
        }

      // UNLOCK ANY PENDING LOCKS
      for (String dbName : getDatabases()) getDatabase(dbName).handleUnreachableNode(nodeLeftName);

      clusterManager.removeServerFromCluster(member, nodeLeftName, removeOnlyDynamicServers);

      OrientDBDistributed ctx = (OrientDBDistributed) serverInstance.getDatabases();

      for (String databaseName : getManagedDatabases()) {
        try {
          if (ctx.getDistributedConfiguration(databaseName).getServerRole(nodeName)
              == ODistributedConfiguration.ROLES.MASTER) {
            reassignClustersOwnership(nodeName, databaseName, false);
          }
        } catch (Exception e) {
          // IGNORE IT
          logger.errorNode(
              nodeName,
              "Cannot re-balance the cluster for database '%s' because the Lock Manager is not"
                  + " available (err=%s)",
              databaseName,
              e.getMessage());
        }
      }

      if (nodeLeftName.equalsIgnoreCase(nodeName))
        // CURRENT NODE: EXIT
        System.exit(1);
    } finally {
      // REMOVE NODE IN DB CFG
      getMessageService().handleUnreachableNode(nodeLeftName);
    }
  }

  @Override
  public DB_STATUS getDatabaseStatus(String iNode, String iDatabaseName) {
    return clusterManager.getDatabaseStatus(iNode, iDatabaseName);
  }

  @Override
  public void setDatabaseStatus(String iNode, String iDatabaseName, DB_STATUS iStatus) {
    clusterManager.setDatabaseStatus(iNode, iDatabaseName, iStatus);
  }

  // Called to notify this server, that a node has been removed from the cluster
  public void onServerRemoved(String nodeName) {
    closeRemoteServer(nodeName);
  }

  // Called when the status of a distributed database changes to online
  public void onDbStatusOnline(String databaseName) {
    final DB_STATUS s = getDatabaseStatus(getLocalNodeName(), databaseName);
    if (s == DB_STATUS.NOT_AVAILABLE) {
      // INSTALL THE DATABASE
      try {
        installDatabase(false, databaseName, false, true);
      } catch (ODistributedLockException lock) {
        setDatabaseStatus(getLocalNodeName(), databaseName, DB_STATUS.NOT_AVAILABLE);
        logger.warn(
            " Failing to acquire lock install database '%s' will retry later ", lock, databaseName);
      }
    }
  }

  // Called when the db config has changed
  public void onDbConfigUpdated(String databaseName, ODocument config) {
    // SEND A DISTRIBUTED MSG TO ALL THE SERVERS
    final Set<String> servers = new HashSet<String>(getActiveServers());
    servers.remove(nodeName);

    if (!servers.isEmpty() && getDatabase(databaseName) != null) {

      final ODistributedResponse dResponse =
          sendRequest(
              databaseName, servers, new OUpdateDatabaseConfigurationTask(databaseName, config));
    }
  }

  public boolean onNodeJoining(final String joinedNodeName) {
    // NOTIFY NODE IS GOING TO BE ADDED. IS EVERYBODY OK?
    for (ODistributedLifecycleListener l : listeners) {
      if (!l.onNodeJoining(joinedNodeName)) {
        return false;
      }
    }
    ((OrientDBDistributed) serverInstance.getDatabases()).connected(new ONodeId(joinedNodeName));
    return true;
  }

  @Override
  public OClusterConfiguration getClusterConfiguration() {
    if (!enabled) return null;

    return clusterManager.getClusterConfiguration();
  }

  @Override
  public String getNodeNameById(int id) {
    return clusterManager.getNodeNameById(id);
  }

  @Override
  public int getNodeIdByName(String node) {
    return clusterManager.getNodeIdByName(node);
  }

  @Override
  public ONodeConfig getNodeConfigurationByUuid(String iNode, boolean useCache) {
    return clusterManager.getNodeConfigurationByUuid(iNode, useCache);
  }

  public void reloadRegisteredNodes() {
    clusterManager.reloadRegisteredNodes();
  }

  public boolean removeNodeFromConfiguration(
      String nodeName,
      String databaseName,
      boolean removeOnlyDynamicServers,
      boolean statusOffline) {
    return clusterManager.removeNodeFromConfiguration(
        nodeName, databaseName, removeOnlyDynamicServers, statusOffline);
  }

  public HazelcastInstance getHazelcastInstance() {
    return clusterManager.getHazelcastInstance();
  }

  @Override
  public ODocument getOnlineDatabaseConfiguration(String databaseName) {
    return clusterManager.getOnlineDatabaseConfiguration(databaseName);
  }

  @Override
  public ODistributedDatabaseImpl getDatabase(String name) {
    return ((OrientDBDistributed) getServerInstance().getDatabases()).getDatabase(name);
  }

  public Set<String> getDatabases() {
    return clusterManager.getDatabases();
  }
}
